package com.udf.common;

import org.apache.flink.table.functions.AggregateFunction;





// 自定义聚合函数
public class CasinoWinAlertAggregate extends AggregateFunction<CasinoWinAlertAggregate.CasinoWinAlertResult, CasinoWinAlertAggregate.CasinoWinAccumulator> {

    // 初始化累加器
    @Override
    public CasinoWinAccumulator createAccumulator() {
        return new CasinoWinAccumulator();
    }

    // 每行数据调用 accumulate 方法
    public void accumulate(CasinoWinAccumulator acc, Integer playerId, Double casinoWin) {
        acc.playerId = playerId;  // 保存 player_id
        acc.sum += casinoWin;

        // 触发报警条件（仅首次超过阈值时触发）
        if (acc.sum >= 1000 ) {
            acc.isAlerted = true;
        }else {
            acc.isAlerted = false;
        }
    }

    public void retract(CasinoWinAccumulator acc, Integer playerId, Double casinoWin) {
        acc.sum -= casinoWin;

    }


    // 返回最终结果
    @Override
    public CasinoWinAlertResult getValue(CasinoWinAccumulator acc) {
        CasinoWinAlertResult result = new CasinoWinAlertResult();
        result.playerId = acc.playerId;
        result.sumCasinoWin = acc.sum;
        result.alertMsg = (acc.sum >= 1000) ?
                String.format("玩家 %d 的累计输赢超过 1000（当前值：%.2f）", acc.playerId, acc.sum) : null;
        return result;
    }

    public static class CasinoWinAccumulator {
        public Integer playerId;
        public Double sum = 0.0;
        public boolean isAlerted = false;
    }

    // 定义聚合结果类型（包含总和及报警信息）
    public static class CasinoWinAlertResult {
        public Integer playerId;
        public Double sumCasinoWin;
        public String alertMsg;
    }
}